package com.reactim.message.controller;

import com.reactim.message.connection.ConnectionInfo;
import com.reactim.message.connection.ConnectionManager;
import com.reactim.message.dto.MessageResponse;
import com.reactim.message.service.BatchMessageService;
import com.reactim.message.service.MessageService;
import com.reactim.message.service.ProtobufMessageService;
import com.reactim.protocol.MessageProto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.ConnectMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;

/**
 * RSocket消息控制器
 * 处理实时消息推送和连接管理
 */
@Slf4j
@Controller
public class MessageRSocketController {

    private final MessageService messageService;
    private final ProtobufMessageService protobufMessageService;
    private final BatchMessageService batchMessageService;
    private final ConnectionManager connectionManager;

    public MessageRSocketController(
            MessageService messageService,
            ProtobufMessageService protobufMessageService,
            BatchMessageService batchMessageService,
            @Qualifier("cachedConnectionManager") ConnectionManager connectionManager) {
        this.messageService = messageService;
        this.protobufMessageService = protobufMessageService;
        this.batchMessageService = batchMessageService;
        this.connectionManager = connectionManager;
    }

    /**
     * 客户端连接建立
     */
    @ConnectMapping("connect")
    public Mono<Void> connect(@Payload UserConnectionInfo connectionInfo, RSocketRequester requester) {
        Long userId = connectionInfo.getUserId();

        // 创建连接信息对象
        ConnectionInfo connInfo = ConnectionInfo.builder()
                .userId(userId)
                .deviceInfo(connectionInfo.getDeviceInfo())
                .clientVersion(connectionInfo.getClientVersion())
                .timestamp(connectionInfo.getTimestamp() != null ? connectionInfo.getTimestamp() : System.currentTimeMillis())
                .build();

        // 使用ConnectionManager存储用户连接
        return connectionManager.storeConnection(userId, requester, connInfo)
                .doOnSuccess(success -> {
                    if (success) {
                        log.info("用户连接建立成功: userId={}, deviceInfo={}", userId, connectionInfo.getDeviceInfo());

                        // 监听连接断开
                        requester.rsocket()
                                .onClose()
                                .doFinally(signal -> {
                                    connectionManager.removeConnection(userId)
                                            .doOnSuccess(removed -> log.info("用户连接断开: userId={}", userId))
                                            .doOnError(error -> log.error("移除用户连接失败: userId={}, error={}", userId, error.getMessage()))
                                            .subscribe();
                                })
                                .subscribe();
                    } else {
                        log.warn("用户连接建立失败: userId={}", userId);
                    }
                })
                .doOnError(error -> log.error("存储用户连接时发生错误: userId={}, error={}", userId, error.getMessage()))
                .then();
    }

    /**
     * 发送消息：客户端通过RSocket发送给服务器（使用Protobuf协议）
     * 用途：客户端主动发送消息给服务器
     * 方向：客户端 → 服务器
     * 场景：用户在聊天界面输入消息，点击发送
     */
    @MessageMapping("message.send")
    public Mono<MessageProto.MessageResponse> sendMessage(@Payload MessageProto.SendMessageRequest protoRequest) {
        // 使用ProtobufMessageService处理Protobuf消息
        return protobufMessageService.sendMessage(protoRequest)
                .doOnSuccess(response -> log.debug("RSocket消息发送成功: {}",
                    response.hasMessageData() ? response.getMessageData().getMessageId() : "unknown"));
    }

    /**
     * 订阅消息推送
     */
    @MessageMapping("message.subscribe")
    public Flux<MessageResponse> subscribeMessages(@Payload MessageSubscribeRequest request) {
        Long userId = request.getUserId();

        // 创建消息推送流
        return Flux.interval(Duration.ofSeconds(1))
                .flatMap(tick -> {
                    // 这里应该从消息队列或缓存中获取待推送的消息
                    // 简化实现，实际应该监听Kafka或Redis的消息推送事件
                    return getUnpushedMessages(userId);
                })
                .doOnSubscribe(subscription -> log.info("用户开始订阅消息: userId={}", userId))
                .doOnCancel(() -> log.info("用户取消订阅消息: userId={}", userId));
    }

    /**
     * 消息已读确认
     */
    @MessageMapping("message.ack")
    public Mono<Void> acknowledgeMessage(@Payload MessageAckRequest request) {
        Long userId = request.getUserId();

        // 更新用户最后活跃时间
        return connectionManager.updateLastActiveTime(userId)
                .then(messageService.markMessageAsRead(request.getMessageId(), userId))
                .doOnSuccess(v -> log.debug("消息已读确认: messageId={}, userId={}",
                    request.getMessageId(), userId))
                .doOnError(error -> log.warn("消息已读确认失败: messageId={}, userId={}, error={}",
                    request.getMessageId(), userId, error.getMessage()));
    }

    /**
     * 批量消息已读确认
     */
    @MessageMapping("message.ack.batch")
    public Mono<Void> acknowledgeMessages(@Payload BatchMessageAckRequest request) {
        Long userId = request.getUserId();

        // 更新用户最后活跃时间
        return connectionManager.updateLastActiveTime(userId)
                .then(messageService.markMessagesAsRead(request.getMessageIds(), userId))
                .doOnSuccess(v -> log.debug("批量消息已读确认: count={}, userId={}",
                    request.getMessageIds().size(), userId))
                .doOnError(error -> log.warn("批量消息已读确认失败: count={}, userId={}, error={}",
                    request.getMessageIds().size(), userId, error.getMessage()));
    }

    /**
     * 批量发送消息
     */
    @MessageMapping("message.send.batch")
    public Mono<com.reactim.message.dto.BatchMessageResponse> sendBatchMessages(@Payload com.reactim.message.dto.BatchSendMessageRequest batchRequest) {
        return batchMessageService.sendBatchMessages(batchRequest)
                .doOnSuccess(response -> log.debug("批量消息发送完成: batchId={}, 成功={}, 失败={}",
                    response.getBatchId(), response.getSuccessCount(), response.getFailureCount()));
    }



    /**
     * 心跳检测
     */
    @MessageMapping("heartbeat")
    public Mono<HeartbeatResponse> heartbeat(@Payload HeartbeatRequest request) {
        Long userId = request.getUserId();

        // 更新用户最后活跃时间
        return connectionManager.updateLastActiveTime(userId)
                .then(Mono.just(HeartbeatResponse.builder()
                        .timestamp(System.currentTimeMillis())
                        .status("OK")
                        .build()))
                .doOnSuccess(response -> log.debug("心跳检测成功: userId={}", userId))
                .doOnError(error -> log.warn("心跳检测时更新活跃时间失败: userId={}, error={}", userId, error.getMessage()));
    }

    /**
     * 推送消息给指定用户（使用Protobuf协议）
     * 这个方法是服务器主动推送消息给客户端
     */
    public Mono<Void> pushMessageToUser(Long userId, MessageResponse message) {
        // 使用ProtobufMessageService创建推送通知
        MessageProto.PushNotification pushNotification = protobufMessageService.createNewMessagePush(message);

        // 使用ConnectionManager发送消息
        return connectionManager.sendMessageToUser(userId, "notification.push", pushNotification)
                .doOnSuccess(v -> log.debug("Protobuf消息推送成功: userId={}, messageId={}", userId, message.getMessageId()))
                .doOnError(error -> log.warn("Protobuf消息推送失败: userId={}, error={}", userId, error.getMessage()));
    }

    /**
     * 推送消息撤回通知
     */
    public Mono<Void> pushMessageRecall(Long userId, String messageId) {
        ClientMessageHandler.MessageRecallNotification notification =
            new ClientMessageHandler.MessageRecallNotification();
        notification.setMessageId(messageId);
        notification.setTimestamp(System.currentTimeMillis());

        return connectionManager.sendMessageToUser(userId, "message.recall", notification)
                .doOnSuccess(v -> log.debug("撤回通知推送成功: userId={}, messageId={}", userId, messageId))
                .doOnError(error -> log.warn("撤回通知推送失败: userId={}, messageId={}, error={}", userId, messageId, error.getMessage()));
    }

    /**
     * 推送已读回执
     */
    public Mono<Void> pushReadReceipt(Long userId, String messageId, Long readByUserId) {
        ClientMessageHandler.ReadReceiptNotification notification =
            new ClientMessageHandler.ReadReceiptNotification();
        notification.setMessageId(messageId);
        notification.setReadByUserId(readByUserId);
        notification.setTimestamp(System.currentTimeMillis());

        return connectionManager.sendMessageToUser(userId, "message.read-receipt", notification)
                .doOnSuccess(v -> log.debug("已读回执推送成功: userId={}, messageId={}", userId, messageId))
                .doOnError(error -> log.warn("已读回执推送失败: userId={}, messageId={}, error={}", userId, messageId, error.getMessage()));
    }

    /**
     * 检查用户是否在线
     */
    public Mono<Boolean> isUserOnline(Long userId) {
        return connectionManager.isUserOnline(userId);
    }

    /**
     * 获取在线用户数量
     */
    public Mono<Long> getOnlineUserCount() {
        return connectionManager.getOnlineUserCount();
    }

    /**
     * 获取用户未推送的消息
     * TODO: 实际实现应该从消息队列或缓存中获取
     */
    private Flux<MessageResponse> getUnpushedMessages(Long userId) {
        // 简化实现，返回空流
        return Flux.empty();
    }

    /**
     * 用户连接信息
     */
    @lombok.Data
    public static class UserConnectionInfo {
        private Long userId;
        private String deviceInfo;
        private String clientVersion;
        private Long timestamp;
    }

    /**
     * 消息订阅请求
     */
    @lombok.Data
    public static class MessageSubscribeRequest {
        private Long userId;
        private Long lastSequence;
        private java.util.List<String> conversationIds;
    }

    /**
     * 消息确认请求
     */
    @lombok.Data
    public static class MessageAckRequest {
        private String messageId;
        private Long userId;
        private Long timestamp;
    }

    /**
     * 批量消息确认请求
     */
    @lombok.Data
    public static class BatchMessageAckRequest {
        private java.util.List<String> messageIds;
        private Long userId;
        private Long timestamp;
    }

    /**
     * 心跳请求
     */
    @lombok.Data
    public static class HeartbeatRequest {
        private Long userId;
        private Long timestamp;
    }

    /**
     * 心跳响应
     */
    @lombok.Data
    @lombok.Builder
    public static class HeartbeatResponse {
        private Long timestamp;
        private String status;
    }



}
